package com.atguigu.flink.state.keyedstate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 黄凯 on 2023/6/20 0020 20:59
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 输出每种传感器不重复的水位值
 * *   去重:
 * *      Set：  flink没有提供Set类型的状态。
 * *      Map<key,value>：使用。把希望去重的数据作为key放入Map
 */
public class Flink08_MapState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {

                    private MapState<Integer, Object> mapState;

                    //初始化方法
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //找到状态的备份存储,取出状态并赋值
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("lastVc", Integer.class, Object.class));
                    }

                    /*
                              KeyedProcessFunction<K, I, O> extends AbstractRichFunction implements RichFunction, Serializable
                                   生命周期方法。
                                      open():  在Task被创建后，执行！ 就是初始化方法！
                                      close():
                                  初始化
                               */
                    @Override
                    public void processElement(WaterSensor value,
                                               KeyedProcessFunction<String, WaterSensor, String>.Context ctx,
                                               Collector<String> out) throws Exception {

                        mapState.put(value.getVc(), "");

                        out.collect(ctx.getCurrentKey() + " 去重后的vc: " + mapState.keys().toString());

                    }
                }
        ).print();

        env.execute();

    }

}
